/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.linkis.cs.contextcache;

import org.apache.linkis.cs.common.entity.enumeration.ContextScope;
import org.apache.linkis.cs.common.entity.enumeration.ContextType;
import org.apache.linkis.cs.common.entity.source.ContextID;
import org.apache.linkis.cs.common.entity.source.ContextKey;
import org.apache.linkis.cs.common.entity.source.ContextKeyValue;
import org.apache.linkis.cs.common.exception.CSErrorException;
import org.apache.linkis.cs.contextcache.cache.ContextCache;
import org.apache.linkis.cs.contextcache.cache.csid.ContextIDValue;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.stream.Collectors;

@Component
public class DefaultContextCacheService implements ContextCacheService {

    private static final Logger logger = LoggerFactory.getLogger(DefaultContextCacheService.class);

    @Autowired private ContextCache contextCache;

    @Override
    public ContextKeyValue put(ContextID contextID, ContextKeyValue csKeyValue)
            throws CSErrorException {

        if (null == contextID || null == csKeyValue) {
            return null;
        }

        logger.info(
                "Start to put contextKey({}) to ContextID({})",
                csKeyValue.getContextKey().getKey(),
                contextID.getContextId());
        ContextIDValue contextIDValue = contextCache.getContextIDValue(contextID);
        ContextKeyValue oldValue = contextIDValue.getContextKeyValueContext().put(csKeyValue);
        logger.info(
                "Finished to put contextKey({}) to ContextID({})",
                csKeyValue.getContextKey().getKey(),
                contextID.getContextId());

        return oldValue;
    }

    @Override
    public ContextKeyValue rest(ContextID contextID, ContextKey csKey) {
        if (null == contextID || null == csKey) {
            return null;
        }
        try {
            ContextIDValue contextIDValue = contextCache.getContextIDValue(contextID);
            return contextIDValue.getContextKeyValueContext().remove(csKey);
        } catch (Exception e) {
            logger.error(
                    String.format(
                            "Failed to rest contextID(%s) of csKey(%s)",
                            contextID.getContextId(), csKey.getKey()),
                    e);
        }
        return null;
    }

    @Override
    public ContextKeyValue get(ContextID contextID, ContextKey csKey) {
        if (null == contextID || null == csKey) {
            return null;
        }
        try {
            ContextIDValue contextIDValue = contextCache.getContextIDValue(contextID);
            return contextIDValue
                    .getContextKeyValueContext()
                    .getContextKeyValue(csKey, csKey.getContextType());
        } catch (Exception e) {
            logger.error(
                    String.format(
                            "Failed to get contextID(%s) of csKey(%s)",
                            contextID.getContextId(), csKey.getKey()),
                    e);
        }
        return null;
    }

    @Override
    public List<ContextKeyValue> getValues(
            ContextID contextID, String keyword, ContextType csType) {
        if (null == contextID || StringUtils.isBlank(keyword)) {
            return null;
        }
        try {
            ContextIDValue contextIDValue = contextCache.getContextIDValue(contextID);
            return contextIDValue.getContextKeyValueContext().getValues(keyword, csType);
        } catch (Exception e) {
            logger.error(
                    String.format(
                            "Failed to getValues contextID(%s) of keyword(%s)",
                            contextID.getContextId(), keyword),
                    e);
        }
        return null;
    }

    @Override
    public List<ContextKeyValue> getAllLikes(
            ContextID contextID, String regex, ContextType csType) {
        if (null == contextID || StringUtils.isBlank(regex)) {
            return null;
        }
        try {
            ContextIDValue contextIDValue = contextCache.getContextIDValue(contextID);
            return contextIDValue.getContextKeyValueContext().getAllLikes(regex, csType);
        } catch (Exception e) {
            logger.error(
                    String.format(
                            "Failed to getAllLikes contextID(%s) of regex(%s)",
                            contextID.getContextId(), regex),
                    e);
        }
        return null;
    }

    @Override
    public List<ContextKeyValue> getAll(ContextID contextID) {

        if (null == contextID) {
            return null;
        }
        try {
            ContextIDValue contextIDValue = contextCache.getContextIDValue(contextID);
            return contextIDValue.getContextKeyValueContext().getAll();
        } catch (Exception e) {
            logger.error(
                    String.format("Failed to getAllByType contextID(%s)", contextID.getContextId()),
                    e);
        }
        return null;
    }

    @Override
    public List<ContextKeyValue> getAllByScope(
            ContextID contextID, ContextScope scope, ContextType csType) {
        if (null == contextID && null == scope) {
            return null;
        }
        try {
            ContextIDValue contextIDValue = contextCache.getContextIDValue(contextID);
            List<ContextKeyValue> valueCSTypeList =
                    contextIDValue.getContextKeyValueContext().getAllValues(csType);
            if (CollectionUtils.isNotEmpty(valueCSTypeList)) {
                return valueCSTypeList.stream()
                        .filter(
                                contextKeyValue ->
                                        scope.equals(
                                                contextKeyValue.getContextKey().getContextScope()))
                        .collect(Collectors.toList());
            }
        } catch (Exception e) {
            logger.error(
                    String.format(
                            "Failed to getAllByScope contextID(%s) of ContextScope(%s) of csType(%s)",
                            contextID.getContextId(), scope, csType),
                    e);
        }
        return null;
    }

    @Override
    public List<ContextKeyValue> getAllByType(ContextID contextID, ContextType csType) {
        if (null == contextID || StringUtils.isBlank(contextID.getContextId())) {
            return null;
        }
        try {
            ContextIDValue contextIDValue = contextCache.getContextIDValue(contextID);
            List<ContextKeyValue> allValues =
                    contextIDValue.getContextKeyValueContext().getAllValues(csType);
            return allValues;
        } catch (Exception e) {
            logger.error(
                    String.format(
                            "Failed to getAllByType contextID(%s) of csType(%s)",
                            contextID.getContextId(), csType),
                    e);
        }
        return null;
    }

    @Override
    public ContextKeyValue remove(ContextID contextID, ContextKey csKey) {
        if (null == contextID || csKey == null) {
            return null;
        }
        try {
            ContextIDValue contextIDValue = contextCache.getContextIDValue(contextID);
            return contextIDValue.getContextKeyValueContext().remove(csKey);
        } catch (Exception e) {
            logger.error(
                    String.format(
                            "Failed to remove contextID(%s) of csKey(%s)",
                            contextID.getContextId(), csKey.getKey()),
                    e);
        }
        return null;
    }

    @Override
    public void removeAll(ContextID contextID) {
        if (null == contextID) {
            return;
        }
        try {
            contextCache.remove(contextID);
        } catch (Exception e) {
            logger.error(
                    String.format("Failed to removeAll contextID(%s)", contextID.getContextId()),
                    e);
        }
    }

    /**
     * TODO This method has poor performance and is not implemented now
     *
     * @param contextID
     * @param scope
     * @param csType
     */
    @Override
    public void removeAll(ContextID contextID, ContextScope scope, ContextType csType) {}

    @Override
    public void removeAll(ContextID contextID, ContextType csType) {
        if (null == contextID) {
            return;
        }
        try {
            ContextIDValue contextIDValue = contextCache.getContextIDValue(contextID);
            contextIDValue.getContextKeyValueContext().removeAll(csType);
        } catch (Exception e) {
            logger.error(
                    String.format(
                            "Failed to removeAll contextID(%s) of csType(%s)",
                            contextID.getContextId(), csType),
                    e);
        }
    }

    @Override
    public void removeByKeyPrefix(ContextID contextID, String preFix) {
        if (null == contextID || StringUtils.isBlank(preFix)) {
            return;
        }
        try {
            ContextIDValue contextIDValue = contextCache.getContextIDValue(contextID);
            contextIDValue.getContextKeyValueContext().removeByKeyPrefix(preFix);
        } catch (Exception e) {
            logger.error(
                    String.format(
                            "Failed to removeByKeyPrefix contextID(%s) of key preFix(%s)",
                            contextID.getContextId(), preFix),
                    e);
        }
    }

    @Override
    public void removeByKeyPrefix(ContextID contextID, String preFix, ContextType csType) {
        if (null == contextID || StringUtils.isBlank(preFix) || null == csType) {
            return;
        }
        try {
            ContextIDValue contextIDValue = contextCache.getContextIDValue(contextID);
            contextIDValue.getContextKeyValueContext().removeByKeyPrefix(preFix, csType);
        } catch (Exception e) {
            logger.error(
                    String.format(
                            "Failed to removeByKeyPrefix contextID(%s) of key preFix(%s) and csyTye(%s)",
                            contextID.getContextId(), preFix, csType.name()),
                    e);
        }
    }
}
